package com.study.spark.scala.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * 窗口函数
 * @author stephen
 * @date 2019-09-25 14:21
 */
object WindowWordCount {

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[2]").setAppName("WindowWordCount")
    // 指定采集周期
    val ssc = new StreamingContext(conf, Seconds(10))
    // 创建SocketInputDStream
    val dstream = ssc.socketTextStream("localhost", 9999)
    // 窗口的大小为采集周期的整数倍，窗口的步长也为采集周期的整数倍
    val lines = dstream.window(Seconds(60),Seconds(20))
    // 进行统计
    val result = lines.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_+_)
    // 打印
    result.print()

    // 启动
    ssc.start()
    ssc.awaitTermination()
  }
}
